package com.sunsharing.dispatch;

import com.sunsharing.common.*;
import com.sunsharing.common.config.ChannelModel;
import com.sunsharing.common.config.Config;
import com.sunsharing.common.config.NodeModel;
import com.sunsharing.common.loadbalance.RandomLoadBalance;
import com.sunsharing.common.register.Register;
import com.sunsharing.common.utils.StringUtils;
import com.sunsharing.dispatch.graph.AF;
import com.sunsharing.dispatch.graph.Graph;
import com.sunsharing.flow.exe.ProcessInstance;
import com.sunsharing.flow.exe.WorkFlowService;
import com.sunsharing.flow.model.ProcessDefinition;
import com.sunsharing.transport.*;
import org.apache.log4j.Logger;

import java.util.*;
import java.util.concurrent.*;

/**
 * Created with IntelliJ IDEA.
 * User: criss
 * Date: 13-4-14
 * Time: 下午9:21
 * To change this template use File | Settings | File Templates.
 */
public class Dispatch {
    static Logger logger = Logger.getLogger(Dispatch.class);
    private String toNode;

    public String getToNode() {
        return toNode;
    }

    public void setToNode(String toNode) {
        this.toNode = toNode;
    }

    static IoHandle clientHandle = new IoHandle() {
        public void run(Context context)
        {
            Dispatch dispatch = new Dispatch();
            dispatch.run(context);
        }
        @Override
        public void doConnected(Channel session) {

        }
    };

    public void run(Context context)
    {
        String nodeId = Config.getInstance().getNodeId();
        //System.out.println(":::"+context);

        //请求并且是第一个节点
        //if(context.isFirstRequest())
            //还没有计算路由，开始计算路由
        //System.out.println(nodeId+":收到消息:"+context.getMsg().msg);



        try{
            if(context.getMsgType() == Constant.MSG_TYPE_REQUEST &&
                    Config.getInstance().isflowService(context.getMsg().serviceId))
            {
                if(!(context.getMsg() instanceof FlowMsg))
                {
                    //启动流程
                    ProcessDefinition pd = WorkFlowService.getFlow(context.getMsg().serviceId);
                    ProcessInstance pro = new ProcessInstance(pd,context);
//                 EventManager.getInstance().removeAllListeners();
//                 FlowEndLisener lisen = new FlowEndLisener();
//                 EventManager.getInstance().addApplicationListener(pro.getDefinition().getEndNode(),lisen);
                    pro.getRootToken().reqId = context.getMsg().reqId;
                    //pro.getRootToken().serviceId = context.getMsg().serviceId;
                    context.getMsg().serviceId = "";
                    pro.getRootToken().addPar(context.getMsg().requests);
                    context.setMsg(pro.getRootToken());
                    pro.start();
                }else
                {
                    //执行流程
                    ((FlowMsg)context.getMsg()).otherNodeExecute();
                }

                return;
            }
            context.addHistory(nodeId);
            //判断服务是否在本机
            if(context.getMsgType() == Constant.MSG_TYPE_REQUEST && context.isLocal())
            {
                //调用服务返回
                logger.info("调用服务----");
                String result = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n" +
                        "<Result>\n" +
                        "<SenderId>XM.GOV.SQ.MZ.RS.yljz. cbryxx</SenderId>\n" +
                        "<ServiceId>XM.GOV.YZ.RS.yljz.cbryxx</ServiceId>\n" +
                        "<Status>000</Status>\n" +
                        "<ErrorMsg></ErrorMsg>\n" +
                        "<Output type=\"string\"><![CDATA[<?xml version=\"1.0\" encoding=\"UTF-8\" ?>\n" +
                        "<Root>\n" +
                        "  <Status>00</Status> \n" +
                        "  <ErrorMsg></ErrorMsg>\n" +
                        "  <Data>\n" +
                        "    <Row>\n" +
                        "      <Item col=”AFZ067”></Item>\n" +
                        "      <Item col=”AAB400”></Item>\n" +
                        "      <Item col=”BAB401”></Item>\n" +
                        "      <Item col=”AFZ001”></Item>\n" +
                        "      <Item col=”AFE006”></Item>\n" +
                        "      <Item col=”AFE005”></Item>\n" +
                        "      <Item col=”BAD005”></Item>\n" +
                        "    </Row>\n" +
                        "  </Data>\n" +
                        "</Root>\n" +
                        "]]>\n" +
                        "  </Output>\n" +
                        "</Result>\n";
                Context restContext = ContextCreate.createResultContext(result,context);
                run(restContext);
                return;
                //work(restContext);

            }else if(context.getMsgType() == Constant.MSG_TYPE_RESULT && context.isLocal())
            {
                List<Channel> list = ConnectClients.searchActiveApiClient(context.getApiNode());
                if(list.size()==0)
                {
                    //TODO 无法返回，请求方已经挂了
                    logger.error("无法找到请求方"+context.getApiNode()+"的连接，挂了，");
                    //ConnectClients.watch();
                    return;
                }else
                {
                    LoadBalance loadBalance = new RandomLoadBalance();
                    int index = loadBalance.getIndex(list.size());
                    list.get(index).send(context);
                }
                return;
            }

            //TODO 跳到下一个节点

            //每到一个点计算路由
            Register reg = Config.getInstance().getRegister();
            String fromNode = Config.getInstance().getNodeId();



            List<NodeModel> toNodes = null;
            if(StringUtils.isBlank(toNode))
            {
                if(context.getMsgType()==Constant.MSG_TYPE_REQUEST)
                {
                    toNodes = Config.getInstance().getNodesByServiceName(context.getMsg().serviceId);
                }else
                {
                    //结果
                    toNodes = new ArrayList<NodeModel>();
                    toNodes.add(Config.getInstance().getNodeModelByNodeId(context.getBeginNode()));
                }
            }else
            {
                toNodes = new ArrayList<NodeModel>();
                toNodes.add(Config.getInstance().getNodeModelByNodeId(toNode));
            }


            List<List<String>> list = new ArrayList<List<String>>();
            for(NodeModel toNode:toNodes)
            {
                AF operation = new AF(new Graph(reg), fromNode, toNode.nodeId);
                list.addAll(operation.getResult());
            }

            if(list.size()==0)
            {
                throw new RuntimeException(fromNode+"到"+toNodes+"无法路由");
            }
            //context.setFromNode(fromNode);



            List<Channel> channels = new ArrayList<Channel>();
            List<String> nextNodes = new ArrayList();
            for(List<String> route:list)
            {
                int index = route.indexOf(fromNode);
                if(index!=-1)
                {
                    String nextNodeId = route.get(index-1);
                    if(!nextNodes.contains(nextNodeId))
                    {
                        nextNodes.add(nextNodeId);
                    }
                }
            }

            List<URL> urls = new ArrayList<URL>();
            for(String nextNode:nextNodes)
            {

                List<ChannelModel> chs= Config.getInstance().getNoByClientChannelModel(nodeId,nextNode);
                if(chs.size()==0)
                {
                    //反向连接
                    channels.addAll(ConnectClients.searchActiveClient(nextNode));
                }else
                {
                    for(ChannelModel ch:chs)
                    {
                        URL tmp = null;
                        if(ch.type.equals(Constant.PROTOCOL_FILE))
                        {
                            //目标的配置
                            tmp = URL.clone(ch.getURL());
                            tmp.putPar(Constant.URL_PAR_TO_NODE_ID,nextNode);
                        }else
                        {
                            tmp =  ch.getURL();
                        }
                        urls.add(tmp);

                    }
                    //tryConnect(urls,nextNode);
                }
            }
            tryConnect(urls,nextNodes);

            List<Channel> chs = ConnectClients.searchActive(nextNodes);
            LoadBalance loadBalance = new RandomLoadBalance();
            int index = loadBalance.getIndex(chs.size());

            //流程异地调用转发
            if(context.getMsg() instanceof FlowMsg)
            {
                ((FlowMsg)context.getMsg()).resetServiceToflow();
            }
            chs.get(index).send(context);

        }catch (Exception e)
        {
            logger.error("消息处理异常",e);

            Context restContext = ContextCreate.createErrorContext(Config.getInstance().getNodeId()+
                    "节点处理出错，错误信息:"+e.getMessage(),context);
            run(restContext);
            return;
        }




    }

    /**
     * errorNode
     */
    static Map errorNode = new HashMap();

    private synchronized static void tryConnect(final List<URL> lists,final List<String> nextNodes)
    {
        //logger.error(Thread.currentThread().getName()+"进入:::");
        if(lists.size()==0)
        {
            return;
        }
        ClientCreater client = new ClientCreater();
        int total = lists.size();
        int error = 0;
        boolean longconnect = false;
        boolean needConnect = false;
        final List<URL> successURL = new ArrayList<URL>();
        for(URL url:lists)
        {
            try{
                if(url.getProtocol().endsWith(Constant.PROTOCOL_MINA) || url.getProtocol().equals(Constant.PROTOCOL_NETTY))
                {
                    longconnect = true;
                }
                Channel tmp = (Channel)ConnectClients.searchClient(url.getProtocol(),url.getIp(),url.getPort());
                if(tmp==null)
                {
                   Long l = (Long)errorNode.get(url.toString());
//                    if(l)
                    //需要连接
                    //List SuccessURL
                    if(l==null || (new Date().getTime()-l)>=30000)
                    {
                        //错误连接超过30秒重新连接
                        client.createClient(url).connect(url, clientHandle);
                        successURL.add(url);
                        needConnect = true;
                        errorNode.remove(url.toString());
                    }else
                    {
                        logger.error(Thread.currentThread().getName()+"已经连接了或者错误连接没有超过30秒不需要重练,"+url);
                    }
                }
            }catch (Exception e)
            {
                logger.error("连接"+url+"出错!!!!!",e);
                errorNode.put(url.toString(),new Date().getTime());
                error++;
            }
        }
        if(!needConnect)
        {
            return;
        }
        if(error==total)
        {
            throw new RuntimeException("无法连接到服务端");
        }
        final boolean flongconnect = longconnect;
        ExecutorService executor = Executors.newSingleThreadExecutor();
        FutureTask<Integer> future =
                new FutureTask<Integer>(new Callable<Integer>()
                {
                    //使用Callable接口作为构造参数
                    public Integer call() {
                        //真正的任务在这里执行，这里的返回值类型为String，可以为任意类型
                        long start = new Date().getTime();
                        while(true)
                        {
                            if(flongconnect)
                            {
                                logger.error("需要确认的连接数:"+successURL.size());
                                boolean allreturn = ConnectClients.allSuccessBack(successURL);
                                if(allreturn)
                                {
                                    logger.error(Thread.currentThread().getName()+":成功连接:"+successURL.size());
                                    return ConnectClients.activeClients.size();
                                }
                            }else
                            {
                                return 1;
                            }

                            logger.error("flongconnenct::::sleep");
                            long end = new Date().getTime();

                            if(end-start>=10000)
                            {
                                return 0;
                            }
                            try{
                                Thread.sleep(200);
                            }catch (Exception e)
                            {

                            }
                        }

                    }
                });
        executor.execute(future);
        try {
            future.get(8000, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            future.cancel(true);
        } catch (ExecutionException e) {
            future.cancel(true);
        } catch (TimeoutException e) {
            future.cancel(true);
        } finally {
            executor.shutdown();
            //logger.error(Thread.currentThread().getName()+"推出:::");
        }



    }



}
